package com.archive.core.processor;


import com.archive.core.config.ArchiveProperties;
import com.archive.core.constants.ArchiveStatementMode;
import com.archive.core.constants.ArchiveStrategy;
import com.archive.core.exception.ArchiveException;
import com.archive.core.executor.*;
import com.archive.core.statistics.ArchiveStatistics;
import com.archive.core.table.ArchiveTable;
import com.archive.core.util.ArchiveUtil;
import com.archive.core.util.SystemClock;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.SQLIntegrityConstraintViolationException;
import java.util.List;
import java.util.Map;

@Slf4j
@Data
public class ArchiveProcessor {

    private DataSource dataSource;
    private DataSource targetDataSource;
    private ArchiveProperties archiveProperties;
    private List<ArchiveTable> archiveTables;
    private Map<String, ArchiveTable> archiveTableMap;
    private ArchiveExecutor archiveExecutor;

    public ArchiveProcessor(ArchiveProperties archiveProperties) {
        this.archiveProperties = archiveProperties;
        this.archiveExecutor = getArchiveExecutor();
    }

    /**
     * 从主表开始归档，这里会检查主表是否开启归档
     */
    public void process(long archiveTimeStamp) {
        ArchiveStatistics archiveStatistics = new ArchiveStatistics();
        try {
            long startTime = SystemClock.now();
            archiveStatistics.start();
            archiveTables.forEach(archiveTable -> {
                try {
                    if (archiveTable.isEnabled() && !ArchiveUtil.checkTimeout(archiveTable, archiveTables, startTime, archiveProperties.getArchiveTimeout())) {
                        process(archiveTable, startTime, archiveTimeStamp, archiveStatistics);
                    }
                } catch (SQLException e) {
                    throw new ArchiveException(e);
                }
            });
        } finally {
            archiveStatistics.end();
            log.info(archiveStatistics.print(archiveTableMap, archiveTimeStamp));
        }

    }

    private void process(ArchiveTable archiveTable, long startTime, long archiveTimeStamp, ArchiveStatistics archiveStatistics) throws SQLException {
        String tableName = archiveTable.getTableName();
        String backTableSuffix = archiveProperties.getArchiveTableSuffix();
        String targetTableName = tableName;
        if (backTableSuffix != null && !backTableSuffix.trim().isEmpty()) {
            targetTableName = tableName + backTableSuffix;
        }
        // 检查并创建归档表
        archiveExecutor.checkArchiveTable(tableName, targetTableName, dataSource.getConnection(),
                targetDataSource == null ? null : targetDataSource.getConnection());
        process(archiveTable, targetTableName, startTime, archiveTimeStamp, archiveStatistics);
    }

    /**
     * 查询并归档数据
     * <p>子级查询条件可能需要父级数据，父级先做查询，查询结果带入子级，到最后一级开始归档，最后一级归档完成后再返回上一级执行归档</p>
     *
     * @param archiveTable     归档表信息
     * @param targetTableName  归档表名
     * @param archiveTimeStamp 归档时间戳
     */
    private void process(ArchiveTable archiveTable, String targetTableName, long startTime, long archiveTimeStamp, ArchiveStatistics archiveStatistics) throws SQLException {
        archiveStatistics.startTable(archiveTable);
        archiveStatistics.queryStart(archiveTable);
        Object maxPrimaryValue = archiveExecutor.queryData(archiveTable, startTime, archiveTimeStamp, null, dataSource.getConnection());
        archiveStatistics.queryEnd(archiveTable);
        while (archiveTable.getData() != null && !archiveTable.getData().isEmpty()) {
            archiveStatistics.increaseRecordCount(archiveTable);
            if (archiveTable.hasRelatedTables()) {
                List<ArchiveTable> relatedTables = archiveTable.getRelatedTables();
                // 先归档关联表
                for (ArchiveTable relatedTable : relatedTables) {
                    process(relatedTable, startTime, archiveTimeStamp, archiveStatistics);
                }
            }
            // 执行归档
            doArchive(archiveTable, targetTableName, archiveStatistics, archiveExecutor);
            // 清空本批次数据
            archiveTable.clear();
            archiveStatistics.queryStart(archiveTable);
            maxPrimaryValue = archiveExecutor.queryData(archiveTable, startTime, archiveTimeStamp, maxPrimaryValue, dataSource.getConnection());
            archiveStatistics.queryEnd(archiveTable);
        }
        archiveStatistics.endTable(archiveTable);
    }

    private ArchiveExecutor getArchiveExecutor() {
        if (ArchiveStrategy.SAME == archiveProperties.getArchiveStrategy()) {
            if (archiveProperties.isUseInsertIntoSelect()) {
                return new SameDBInsertIntoSelectArchiveExecutor(this);
            }
            return new SameDBArchiveExecutor(this);
        } else {
            return new BackupDBArchiveExecutor(this);
        }
    }

    /**
     * 执行归档
     *
     * @param archiveTable    归档表信息
     * @param targetTableName 归档表名
     */
    public void doArchive(ArchiveTable archiveTable, String targetTableName, ArchiveStatistics archiveStatistics, ArchiveExecutor archiveExecutor) throws SQLException {
        Connection connection = null;
        Connection targetConnection = null;
        boolean autoCommit = false;
        boolean targetAutoCommit = false;
        long start = SystemClock.now();

        try {
            connection = dataSource.getConnection();
            autoCommit = connection.getAutoCommit();
            connection.setAutoCommit(false);

            if (archiveProperties.getArchiveStrategy() == ArchiveStrategy.BACKUP) {
                targetConnection = targetDataSource.getConnection();
                targetAutoCommit = targetConnection.getAutoCommit();
                targetConnection.setAutoCommit(false);
            }

            // 预编译、一般方式执行
            if (archiveProperties.getStatementMode() == ArchiveStatementMode.NORMAL) {
                archiveExecutor.doArchiveNormal(archiveTable, targetTableName, connection, targetConnection, archiveStatistics);
            } else {
                archiveExecutor.doArchivePrepared(archiveTable, targetTableName, connection, targetConnection, archiveStatistics);
            }

            // 先插入，再删除，后面会有主键重复处理（不同库先删除，再插入可能会丢数据）
            if (targetConnection != null) {
                targetConnection.commit();
            }
            connection.commit();
            log.info("table {} doArchive cost time：{}ms", archiveTable.getTableName(), SystemClock.now() - start);
        } catch (Exception e) {
            if (connection != null) {
                connection.rollback();
            }
            if (targetConnection != null) {
                targetConnection.rollback();
            }

            if (e.getCause() instanceof SQLIntegrityConstraintViolationException) {
                // 主键重复处理，不同库归档，Connection提交不是原子操作，可能导致插入主键重复
                archiveExecutor.handleDuplicateKeyException(archiveTable, targetTableName, connection, targetConnection, archiveStatistics);
                // 开始时间记录后抛出异常，这里要结束
                archiveStatistics.executeInsertEnd(archiveTable);
                return;
            }

            throw e;
        } finally {
            if (connection != null) {
                connection.setAutoCommit(autoCommit);
            }
            if (targetConnection != null) {
                targetConnection.setAutoCommit(targetAutoCommit);
            }
            ArchiveUtil.closeConnection(connection);
            ArchiveUtil.closeConnection(targetConnection);
        }
    }

    public void setOriginConnectionProvider(OriginConnectionProvider originConnectionProvider) {
        archiveExecutor.setOriginConnectionProvider(originConnectionProvider);
    }

}
